package net.onething.dc.hdfs2bos;

import java.io.IOException;
import java.util.concurrent.*;

import com.baidubce.services.bos.BosClient;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class HdfsToBos {

    private static final Logger log = LoggerFactory.getLogger(HdfsToBos.class);

    private ConfigReader configReader;
    private BlockingQueue<FileToBosTask> taskBlockingQueue;
    private BosClient bosClient;

    public HdfsToBos(ConfigReader configReader,
                     BlockingQueue<FileToBosTask> taskBlockingQueue,
                     BosClient bosClient) {
        this.configReader = configReader;
        this.taskBlockingQueue = taskBlockingQueue;
        this.bosClient = bosClient;
    }

    private void submitTask(FileToBosTask task) throws Exception {
        if (null == task) {
            throw new IllegalArgumentException("can not submit a null task.");
        }

        if (null == this.taskBlockingQueue) {
            throw new NullPointerException("can not submit a task to null "
                    + "blocking queue.");
        }
        this.taskBlockingQueue.put(task);
    }

    private void scanHdfsMember(Path hdfsPath, FileSystem hdfsFs) throws Exception {
        FileStatus targetPathStatus = hdfsFs.getFileStatus(hdfsPath);
        if (targetPathStatus.isFile()) {
            this.submitTask(this.buildHdfsFileToCosTask(targetPathStatus));
            return;
        }
        FileStatus[] memberArray = hdfsFs.listStatus(hdfsPath);
        for (FileStatus member : memberArray) {
            this.submitTask(this.buildHdfsFileToCosTask(member));
            if (member.isDirectory()) {
                scanHdfsMember(member.getPath(), hdfsFs);
            }
        }
    }

    private FileToBosTask buildHdfsFileToCosTask(FileStatus fileStatus) {
        if (null == fileStatus) {
            log.error("parameter file status is null.");
            return null;
        }
        FileToBosTask task = null;
        try {
            task = new FileToBosTask(
                    this.configReader,
                    this.bosClient, fileStatus,
                    this.configReader.getHdfsFs(),
                    CommonHdfsUtils.convertTobosPath(configReader, fileStatus.getPath()).toString());
        } catch (IOException e) {
            log.error("build a hdfsFileToCosTask for " + fileStatus.toString()
                    + " failed. exception: " + e.getMessage());
            return null;
        }
        return task;
    }

    public void run() {
        try {
            FileSystem hdfsFS = configReader.getHdfsFs();
            String srcPath = configReader.getSrcHdfsPath();
            this.scanHdfsMember(new Path(srcPath), hdfsFS);
        } catch (Exception e) {
            log.error("Scanning hdfs files occurs an exception.", e);
        }
    }
}
